package com.feiwei

import com.feiwei.SensorReading

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala._

object day1_TableApiTest {


  def main(args: Array[String]): Unit = {


    val en = StreamExecutionEnvironment.getExecutionEnvironment


    val text: DataStream[String] = en.readTextFile("E:\\repository\\company\\myself\\flink-learning\\flink-learning-demo\\src\\main\\resources\\sensor.txt")
    val res = text.map(data=>{
      val arr=data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

   // res.print()
    //  en.execute()

    val  tableEnv=StreamTableEnvironment.create(en)


    val table: Table = tableEnv.fromDataStream(res)
    tableEnv.createTemporaryView("dateTable",table)

    val sql="select id,temperature from dateTable "
   val sqlres: Table = tableEnv.sqlQuery(sql)
    val table2= table.select("id,temperature")
      .filter("id=='sensor_1'")


    table2.toAppendStream[(String,Double)].print("result")

   sqlres.toAppendStream[(String,Double)].print("sqlresult")
    en.execute()
  }
}
